package Streaming;

import java.lang.reflect.Parameter;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @Author: 蔡翔
 * @Date: 2019/10/5 18:43
 * @Version 1.0
 *
 * 并行相关的知识点
 */
public class StreamDemo5 {
    public static void main(String[] args) {
        // 现在是串行，一行一行打印的。
        // 调用parallel() 参数并行流。
        //IntStream.range(1,100).parallel().peek(StreamDemo5::debug).count();

        // demo1
        // 现在实现这样一个效果：现在有两步操作：第一步想要并行操作，第二步想要串行操作。
        // 同时调用 parallel和sequential 以最后一次为准。 所以下面的demo 是串行的
        /*IntStream.range(1,100)
                //调用parallel产生并行流。
                .parallel().peek(StreamDemo5::debug)
                //调用sequential 产生 串行流
                .sequential().peek(StreamDemo5::debug2)
                .count();*/

        // demo2
        // 并行流使用的线程池：ForkJoinPool.commonPool
        // 默认的线程数是，当前机器的cpu个数
        // 使用这个属性可以修改默认的线程数(现在并行的线程是5)
        /*System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","5");
        IntStream.range(1,100).parallel().peek(StreamDemo5::debug).count();*/

        // demo3
        //如果所有的并行流 都用一个线程池的话 就会造成线程的堵塞 ，，可以使用自己的线程池（防止任务被阻塞）
        //目的： 把并行任务 放到自己的线程池里面做。 ForkJoinPool是一个守护线程
        ForkJoinPool pool = new ForkJoinPool(20);
        //把任务提交给 线程池
        pool.submit(()->IntStream.range(1,100).parallel().peek(StreamDemo5::debug).count());
        pool.shutdown();

        // 让main 线程等待 ForkJoinPool线程池里面的线程。
        synchronized (pool){
            try {
                pool.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void debug(int i){
        System.out.println(Thread.currentThread().getName()+"debug:"+i);
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void debug2(int i){
        System.err.println(Thread.currentThread().getName()+"debug2:"+i);
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
